package org.example.zookeeper.test;

import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.exception.ZkMarshallingError;
import org.I0Itec.zkclient.serialize.ZkSerializer;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;
import org.example.zookeeper.config.ZkProperties;

import java.util.List;

public class ZkClientTester implements ZookeeperTester {
    @Override
    public void test(ZkProperties zkConfig) {
        ZkClient client = new ZkClient(zkConfig.getServer(), zkConfig.getSessionTimeout(), zkConfig.getConnectionTimeout());
        client.setZkSerializer(new ZkSerializer() {
            @Override
            public byte[] serialize(Object data) throws ZkMarshallingError {
                return (byte[]) data;
            }

            @Override
            public Object deserialize(byte[] bytes) throws ZkMarshallingError {
                return bytes;
            }
        });
        try {
            String path = "/zk-client";
            // IllegalArgumentException: Path must start with / character
            // createNode(client, "start", CreateMode.PERSISTENT);
            deleteNode(client, path);
            deleteNode(client, path);
            exists(client, path);

            createNode(client, path, CreateMode.PERSISTENT);
            createNode(client, path + "/c1", CreateMode.PERSISTENT);
            createNode(client, path + "/c1/c11", CreateMode.PERSISTENT);
            createNode(client, path + "/c1/c12", CreateMode.PERSISTENT);
            createNode(client, path + "/e1", CreateMode.EPHEMERAL);

            getChildren(client, path);
            getChildren(client, path + "/temp");
            getChildren(client, path + "/c1");

            setData(client, path + "/c1", "default".getBytes());

            getDate(client, path + "/c1");
            getDate(client, path + "/c2");

            setData(client, path + "/c1", null);
            setData(client, path + "/c2", "hello".getBytes());

            getDate(client, path + "/c1");

            print("fin");
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (client != null) {
                client.close();
            }
        }
    }

    private void exists(ZkClient client, String path) {
        boolean b = client.exists(path);
        print("exists " + b);
    }

    private void deleteNode(ZkClient client, String path) {
        boolean b = client.deleteRecursive(path);
        System.out.println(b);
    }

    private void createNode(ZkClient client, String path, CreateMode mode) {
        try {
            String v = client.create(path, "default".getBytes(), mode);
            print("created " + v);
        } catch (Exception e) { // KeeperException.NodeExistsException
            print("create node error: " + e.getMessage());
        }
    }

    private void getDate(ZkClient client, String path) {
        try {
            Stat stat = new Stat();
            byte[] bytes = client.readData(path, stat);
            if (bytes != null) {
                print("data is " + new String(bytes));
            } else {
                print("not data found");
            }
            print("data stat: " + stat);
        } catch (Exception e) { // ZkMarshallingError
            print("node not found");
        }
    }

    private void setData(ZkClient client, String path, byte[] data) {
        try {
            client.writeData(path, data);
        } catch (Exception e) {
            print("[warning] node not found");
        }
    }

    private void getChildren(ZkClient client, String path) {
        try {
            List<String> children = client.getChildren(path);
            print("children size: " + children.size() + " children is: " + children);
        } catch (Exception e) {
            print("[warning] parent node not found");
        }
    }
}
